package org.kth.dks.dks_marshal;

import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.kth.dks.dks_comm.DKSRef;

/**
 * Runs inside a thread, supposedly one per DKSNode. Keeps a synchronized buffer, and
 * dispatches messages to the right DKSNode handler.
 * <p>Title: DKS</p>
 * <p>Description: DKS Middleware</p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: KTH-IMIT/SICS</p>
 * @author not attributable
 * @version 1.0
 */

public class ObjectAdapter implements Runnable {
	
    private Logger log = Logger.getLogger(ObjectAdapter.class);
	private boolean stopped = false;
	private List msgQueue = Collections.synchronizedList(new LinkedList());
	//private final List msgQueue = new LinkedList();
	private final long MAXQUEUE = 1000000; // Max entries in the msgQueue
	private Map handlerMap = new HashMap(); // maps a DKSMessage type to a handler object+method
	
	// values inside handlerMap are of type HandlerPair
	class HandlerPair {
		private final Method method;
		private  final Object handlerObject;
		public HandlerPair(Method m, Object o) { method=m; handlerObject=o; }
		public Method getMethod() { return method; }
		public Object getHandlerObject() { return handlerObject; }
	}
	
	class MessagePair {
		private final DKSMessage msg;
		private final DKSRef src;
		public MessagePair(DKSMessage m, DKSRef s) { msg=m; src=s; }
		public DKSMessage getMsg() { return msg; }
		public DKSRef getSrc() { return src; }
	}
	
	public ObjectAdapter() {
	}
	
	/**
	 * Takes a bite off the msgQueue and dispatches it to the right DKSNode handler
	 * Stops if stop() is called.
	 */
	public void run() {
		while (!stopped) {
			MessagePair mp = getMessagePair();
			if (mp!=null) {
				callMessageHandler(mp.getMsg(), mp.getSrc());
			}
		}
		msgQueue.clear(); // garbage collect everything
	}
	
	/**
	 * Stop the running thread (for termination).
	 */
	public void stop() {
		stopped=true;
	}
	
	/**
	 * Synchronized method to put a message together with its sender in the queue
	 * @param m DKSMessage marshalled message
	 * @param s DKSRef reference to the sending node
	 * @return boolean true if successful, otherwise false.
	 */
	public boolean putMessagePair(DKSMessage m, DKSRef s) {
		synchronized (msgQueue) {
			while (msgQueue.size()>=MAXQUEUE) {
				try {
					msgQueue.wait();
				} catch(Exception ex) 
				{ 
					log.error( ex+""); 
				}
			}
			boolean result = msgQueue.add(new MessagePair(m, s));
			if (msgQueue.size()==1) 
				msgQueue.notify();
			return result;
		}
	}
	
	/**
	 * Synchronized get method that returns a DKSMessage and the assoicated sender of the message
	 * @return MessagePair containing the message and the sender, null if interrupted (on exit)
	 */
	public MessagePair getMessagePair() {
		MessagePair mp = null;
		synchronized (msgQueue) {
			while (msgQueue.isEmpty()) {
				try {
					msgQueue.wait();
				} catch (InterruptedException ex) {
					stop();
					return null;
				}
			}
			if (msgQueue.size()==MAXQUEUE)
				msgQueue.notifyAll();
			mp = (MessagePair) msgQueue.remove(0);
		}
		return mp;
	}
	
	/**
	 * Maps a message class, e.g. "dks_marshal.LookupMsg", to a method of a class Z and an object of type Z
	 * @param messageClassZ String represeting the DKSMessage class
	 * @param handlerClassZ String representing the class of the method
	 * @param handlerMethodZ String method name
	 * @param handlerObject Object object of type handlerClassZ which has method handlerMethodZ
	 */
	public boolean addMsgHandler(String messageClassZ,
			String handlerClassZ,
			String handlerMethodZ,
			Object handlerObject) {
		try {
			
			Class cdksref = Class.forName("org.kth.dks.dks_comm.DKSRef");
			Class msgClass = Class.forName(messageClassZ);
			Class handlerClass = Class.forName(handlerClassZ);
			Method handlerMethod = handlerClass.getDeclaredMethod(handlerMethodZ,
					new Class[] {cdksref, msgClass});
			handlerMap.put(msgClass, new HandlerPair(handlerMethod, handlerObject));
			return true;
		} catch (Exception ex) {
			log.error(
					"Class not found, could not install unmarshaller (MessageClass="+messageClassZ+") (HandlerClass="+
					handlerClassZ+") (HandlerMethod="+handlerMethodZ+")\n"+ex);
			return false;
		}
	}
	
	/**
	 * Takes a narrowed DKSMessage and dispatches it to a handler
	 * @param dksMsg DKSMessage narroved message
	 */
	public boolean callMessageHandler(DKSMessage dksMsg, DKSRef source) {
		try {
			Class messageClass = dksMsg.getClass();
			
			if (handlerMap.get(messageClass)==null)
				return false;
			HandlerPair hp = (HandlerPair) handlerMap.get(messageClass);
			Method handlerMethod = hp.getMethod();
			Object handlerObject = hp.getHandlerObject();
			
			log.debug("Invoking(msg:"+dksMsg.getClass()+", src:"+source+")");
			
			handlerMethod.invoke(handlerObject, new Object[] {source, dksMsg});
			
		} catch (Exception ex) {
			ex.printStackTrace();
			return false;
		}
		return true;
	}
	
}
